package top.houry.netty.barrage.config;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;
import top.houry.netty.barrage.bo.BarrageMsgBo;
import top.houry.netty.barrage.service.impl.BarrageReceiveMsgServiceImpl;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * redis订阅频道处理类
 * 弹幕服务监听到频道有新消息时触发
 */
//@Component
@Slf4j
public class RedisListenerHandler extends MessageListenerAdapter {
    @Value("${redis.channel.msgToAll}")
    private String msgToAll;
    private BarrageNettyConfigProperties nettyConfigProperties;

    @Autowired
    public void setNettyConfigProperties(BarrageNettyConfigProperties nettyConfigProperties) {
        this.nettyConfigProperties = nettyConfigProperties;
    }

    private RedisTemplate<String, String> redisTemplate;
    private BarrageReceiveMsgServiceImpl barrageReceiveMsgService;
    @Autowired
    public void setBarrageReceiveMsgService(BarrageReceiveMsgServiceImpl barrageReceiveMsgService) {
        this.barrageReceiveMsgService = barrageReceiveMsgService;
    }

    @Autowired
    public void setRedisTemplate(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     *  频道有新消息时触发
     * 订阅之后，频道收到的消息
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        byte[] body = message.getBody();
        byte[] channel = message.getChannel();
        String rawMsg;
        String topic;
        try {
            rawMsg = redisTemplate.getStringSerializer().deserialize(body);
            topic = redisTemplate.getStringSerializer().deserialize(channel);
            log.info("收到了订阅主题发布的消息：");
            log.info("Received raw message from topic:" + topic + ", raw message content：" + rawMsg);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return;
        }
        if (msgToAll.equals(topic)) {
            log.info("Send message to all users:" + rawMsg);
            BarrageMsgBo barrageMsgBo = JSONUtil.toBean(rawMsg, BarrageMsgBo.class);
            if (checkNotMySelf(barrageMsgBo)){
                // 发布的时候，需要把自己的唯一标识给带上
                // 订阅的时候，如果发现是自己的唯一标识，则不处理，否则进行转发逻辑
                // 获取自己的ip:port 并从消息里获取唯一标识消息 判断是否一致决定是否转发
                barrageReceiveMsgService.notifyOtherUser(barrageMsgBo.getVideId(),barrageMsgBo);
            }
        } else {
            log.warn("No further operation with this topic!");
        }
    }

    /**
     * 检查当前的消息是不是自己发布的
     * true ： 不是自己发的
     * false : 是自己发的
     */
    private boolean checkNotMySelf(BarrageMsgBo barrageMsgBo) {
        try {
            InetAddress localHost = InetAddress.getLocalHost();
            return !(barrageMsgBo.getIp().equals(localHost.getHostAddress()) && barrageMsgBo.getPort().equals(nettyConfigProperties.getServerPort()));
        } catch (UnknownHostException e) {
            throw new RuntimeException("获取本地 ip 地址失败");
        }
    }
}
